package com.steven.logserver.handler;

import com.steven.logserver.config.ZookeeperConfig;
import com.steven.logserver.mq.DirectMsgProducer;
import com.steven.logserver.protobuf.LogServerSendRespProto;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.apache.curator.framework.recipes.atomic.AtomicValue;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;
import java.net.UnknownHostException;

/**
 * 服务端处理器（用于测试调试，不做starter模块的功能）
 *
 * TIPS：如果Handler处理器有一些长时间的业务处理，可以交给taskQueue异步处理
 *
 * @author qinxh
 * @date 2020/08/13
 */
public class MyServerHandler extends ChannelInboundHandlerAdapter {

    private static final Logger logger = LoggerFactory.getLogger(MyServerHandler.class);

    private DistributedAtomicInteger atomicInteger;

    private DirectMsgProducer producer;

    public MyServerHandler(Integer bindPort, String clientNumberPath, DirectMsgProducer producer) throws UnknownHostException {

        this.producer = producer;
        // curator的实现分布式计数器实现当前服务端连接客户端的统计。其分布式计数器对应的节点名为: 自定义的clientNumberPath + netty服務端地址
        // 实现服务端连接客户端个数的统计
        String path = clientNumberPath + InetAddress.getLocalHost().getHostAddress() + ":" + bindPort;
        atomicInteger = new DistributedAtomicInteger(ZookeeperConfig.getClient(), path, ZookeeperConfig.getRetryPolicy());
    }

    /**
     * 判断有客户端连接进来，需要将该netty服务端对应的zk节点信息中的客户端连接个数加一
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public synchronized void channelActive(ChannelHandlerContext ctx) throws Exception {

        AtomicValue<Integer> increment = atomicInteger.increment();
        logger.info("连接成功!客户端数量最新值: {}", increment.postValue());
    }

    /**
     *  实现ChannelInboundHandlerAdapter处理器的读事件fireChannelRead()
     *
     * @param ctx       ChannelHandlerContext上下文对象，就可以拿到channel、pipeline等对象，从而进行读写等操作
     * @param msg       客户端发送过来的消息
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {

        // 每个EventLoopGroup里包括一个或多个EventLoop，每个EventLoop中维护一个Selector实例
        ctx.channel().eventLoop().execute(() -> {
            // 获取客户端发送过来的消息。通过mq异步处理
            producer.sendMsg(msg);
        });
    }

    /**
     * 实现ChannelInboundHandlerAdapter处理器的读完成事件fireChannelReadComplete()
     *
     * @param ctx       ChannelHandlerContext上下文对象
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {

        //发送Proto格式消息给客户端
        LogServerSendRespProto.LogServerSendResp.Builder builder = LogServerSendRespProto.LogServerSendResp.newBuilder();
        builder.setCode(666666);
        builder.setMessage("The server has received the message");
        ctx.writeAndFlush(builder.build());
    }

    /**
     *  判断客户端连接断开，需要将该netty服务端对应的zk节点信息中的客户端连接个数减一
     *
     * @param ctx           ChannelHandlerContext上下文对象
     * @param cause         Throwable
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

        AtomicValue<Integer> increment = atomicInteger.decrement();
        logger.info("客户端关闭!客户端数量最新值: {}", increment.postValue());

        ctx.close();
    }
}
